package com.swhy

import java.net.URI
import java.text.SimpleDateFormat
import java.util.{Date, UUID}

import com.utils._
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
import org.codehaus.jackson.map.ObjectMapper
import java.util
import java.util

import net.minidev.json.JSONObject
import net.minidev.json.parser.JSONParser
import org.slf4j.LoggerFactory

import scala.collection.JavaConversions._
import scala.collection.mutable
import scala.util.parsing.json.JSON


object ChgeCustInfo {
  def main(args: Array[String]): Unit = {
    val logger = LoggerFactory.getLogger(this.getClass)

    if (args.length != 3) {
      println(
        """
          |com.swhy.CompanyInfo
          |params:
          | InputDate    刷数日期
          | OutputPath   输出路径
          | SplitNum     文件生成数量
        """.stripMargin)
      sys.exit()
    }
    val conf = new Configuration()
    val Array(inputDt, outputPath, splitNum) = args
    val sparkConf = new SparkConf().setAppName(this.getClass.getName)
    val sc = new SparkContext(sparkConf)
    val hiveContext = new HiveContext(sc)
    //改下方的sql为需要装换成jason的sql
    val allcompany =
      s"""
         |select
         |A.ecif_cd,
         |A.cust_no,
         |A.capi_acc,
         |A.cust_name,
         |A.sex,
         |A.cust_char_cd,
         |A.ocat_dt,
         |A.birth_dt,
         |A.mob,
         |A.cont_addr,
         |A.latest_risk_grad,
         |A.cust_grad,
         |A.cust_type,
         |cast(A.bjs_flag as int) bjs_flag,
         |cast(A.marg_flag as int) marg_flag,
         |cast(A.hgt_flag as int) hgt_flag,
         |cast(A.sgt_flag as int) sgt_flag,
         |cast(A.opt_flag as int) opt_flag,
         |cast(A.secu_accet_flag as int) secu_accet_flag,
         |cast(A.pro_investor_flag as int) pro_investor_flag,
         |cast(A.effect_flag as int) effect_flag,
         |A.mgr_serv,
         |A.mgr_dev,
         |A.mgr_sign,
         |A.tm,
         |nvl(B.indx_val,'[]') as social_relation
         |from
         |app_data.app_dao_acct_info_d_q A
         |left join
         |app_data.app_dao_acct_ind_q B
         |on A.capi_acc = B.id
         |and B.id_type='a0101'
         |and B.indx_no = 'social_relation'
         |where A.biz_date=${inputDt}
      """.stripMargin
    print(allcompany)
    val frame = hiveContext.sql(allcompany)
    val result = frame.rdd.map(x => {
      var cominfo = new StringBuilder()
      val ecif_cd = x.getAs[String]("ecif_cd")
      val cust_no = x.getAs[String]("cust_no")
      val capi_acc = x.getAs[String]("capi_acc")
      val cust_name = x.getAs[String]("cust_name")
      val sex = x.getAs[String]("sex")
      val cust_char_cd = x.getAs[String]("cust_char_cd")
      val ocat_dt = x.getAs[String]("ocat_dt")
      val birth_dt = x.getAs[String]("birth_dt")
      val mob = x.getAs[String]("mob")
      val cont_addr = x.getAs[String]("cont_addr")
      val latest_risk_grad = x.getAs[String]("latest_risk_grad")
      val cust_grad = x.getAs[String]("cust_grad")
      val cust_type = x.getAs[String]("cust_type")
      //获取json格式字符串
      val social_relation = x.getAs[String]("social_relation")
      val bjs_flag = x.getAs[Integer]("bjs_flag")
      val marg_flag = x.getAs[Integer]("marg_flag")
      val hgt_flag = x.getAs[Integer]("hgt_flag")
      val sgt_flag = x.getAs[Integer]("sgt_flag")
      val opt_flag = x.getAs[Integer]("opt_flag")
      //val fnd_flag = x.getAs[Integer]("fnd_flag")
      val secu_accet_flag = x.getAs[Integer]("secu_accet_flag")
      val pro_investor_flag = x.getAs[Integer]("pro_investor_flag")
      val effect_flag = x.getAs[Integer]("effect_flag")
      val mgr_serv = x.getAs[String]("mgr_serv")
      val mgr_dev = x.getAs[String]("mgr_dev")
      val mgr_sign = x.getAs[String]("mgr_sign")
      val tm = x.getAs[Date]("tm")

      val _id = new ID()
      val index = new Index()
      val companyInfo = new ChgeCustInfo()
      //注释
      //123
      /*val social_relationList = new util.ArrayList[SocialRelation]()
      logger.info("================social_relation是：" + social_relation)
      if (social_relation != null && !social_relation.equals("") && !social_relation.equals("N")){
        logger.info("================是否进入for......................")
        val social_relation1 = social_relation.replace("[","")
        val social_relation2 = social_relation1.replace("]","")
        var arraySocial = social_relation2.split(";")

        for (i <- 0 to arraySocial.length-1){
          val x = arraySocial(i)
          //字符串转json对象
          val JSONObject = new JSONParser()
          val jsonObj: JSONObject = JSONObject.parse(x).asInstanceOf[JSONObject]

          val capiAcc = jsonObj.get("capi_acc").toString
          val socialRelation = jsonObj.get("social_relation").toString
          val acctNameRela = jsonObj.get("acct_name_rela").toString
          val acctRela = jsonObj.get("acct_rela").toString
          val relaCfm = jsonObj.get("rela_cfm").toString
          val bizDm = Integer.parseInt(jsonObj.get("biz_dm").toString)
          val social = new SocialRelation()
          //转换为date类型
          if (jsonObj.get("cfm_dm") != null){
          val cfmDm = jsonObj.get("cfm_dm").toString
          val dateformat = new SimpleDateFormat("yyyy-MM-dd")
            val date = dateformat.parse(cfmDm);
            social.setCfmDm(date)
          }
          /*val jsonOBJ :JSONObject  = JSON.parseObject(b)
          val capiAcc : String  = jsonOBJ.getString("capi_acc")
          val socialRelation : String  = jsonOBJ.getString("social_relation")
          val acctNameRela : String  = jsonOBJ.getString("acct_name_rela")
          val acctRela : String  = jsonOBJ.getString("acct_rela")
          val relaCfm : String  = jsonOBJ.getString("rela_cfm")
          val bizDm : Integer  = Integer.parseInt(jsonOBJ.getString("biz_dm"))
          val cfmDm : String  = jsonOBJ.getString("cfm_dm")
          val dateformat = new SimpleDateFormat("yyyy-MM-dd")
          val date = dateformat.parse(cfmDm)*/
          social.setCapiAcc(capiAcc)
          social.setSocialRelation(socialRelation)
          social.setAcctNameRela(acctNameRela)
          social.setAcctRela(acctRela)
          social.setRelaCfm(relaCfm)
          social.setBizDm(bizDm)
          social_relationList.add(social)
        }
      }*/


      _id.set_id(UUID.randomUUID().toString)
      index.setIndex(_id)
      companyInfo.setEcifCd(ecif_cd)
      companyInfo.setCustNo(cust_no)
      companyInfo.setCapiAcc(capi_acc)
      companyInfo.setCustName(cust_name)
      companyInfo.setSex(sex)
      companyInfo.setCustCharCd(cust_char_cd)
      companyInfo.setOcatDt(ocat_dt)
      companyInfo.setBirthDt(birth_dt)
      companyInfo.setMob(mob)
      companyInfo.setContAddr(cont_addr)
      companyInfo.setLatestRiskGrad(latest_risk_grad)
      companyInfo.setCustGrad(cust_grad)
      companyInfo.setCustType(cust_type)
      companyInfo.setSocialRelation(social_relation)
      companyInfo.setBjsFlag(bjs_flag)
      companyInfo.setMargFlag(marg_flag)
      companyInfo.setHgtFlag(hgt_flag)
      companyInfo.setSgtFlag(sgt_flag)
      companyInfo.setOptFlag(opt_flag)
      //companyInfo.setFndFlag(fnd_flag)
      companyInfo.setSecuAccetFlag(secu_accet_flag)
      companyInfo.setProInvestorFlag(pro_investor_flag)
      companyInfo.setEffectFlag(effect_flag)
      companyInfo.setMgrServ(mgr_serv)
      companyInfo.setMgrDev(mgr_dev)
      companyInfo.setMgrSign(mgr_sign)
      companyInfo.setTm(tm)

      val mapper = new ObjectMapper()

      val indexJson = mapper.writeValueAsString(index) //企业自增编码json串
      val companyInfoJson = mapper.writeValueAsString(companyInfo) //企业信息串

      cominfo = cominfo.append(indexJson + "\n" + companyInfoJson) //json结果串
      cominfo.toString()
    })
    val path = new Path(outputPath) //创建文本输出路径
    val fs = FileSystem.get(URI.create(outputPath), conf)
    if (fs.exists(path)) {
      fs.delete(path, true)
    }
    result.coalesce(splitNum.toInt,true).saveAsTextFile(outputPath)
    sc.stop()
  }
}
